package com.googlecode.cswish.struts.spring;

import java.io.Serializable;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.sql.Connection;
import java.sql.Date;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import javax.persistence.Embeddable;
import javax.persistence.Embedded;
import javax.persistence.EntityManager;
import javax.persistence.PersistenceContext;

import net.sf.ehcache.CacheException;
import net.sf.ehcache.CacheManager;
import net.sf.ehcache.Ehcache;
import net.sf.ehcache.search.Attribute;
import net.sf.ehcache.search.Direction;
import net.sf.ehcache.search.Query;
import net.sf.ehcache.search.Result;
import net.sf.ehcache.search.Results;
import net.sf.ehcache.search.attribute.AttributeExtractor;
import net.sf.ehcache.search.expression.Criteria;
import net.sf.ehcache.store.Store;
import net.sf.ehcache.util.Timestamper;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.hibernate.Filter;
import org.hibernate.Session;
import org.hibernate.SessionFactory;
import org.hibernate.cache.spi.CacheKey;
import org.hibernate.engine.spi.FilterDefinition;
import org.hibernate.engine.spi.SessionFactoryImplementor;
import org.hibernate.event.spi.EventSource;
import org.hibernate.internal.FilterConfiguration;
import org.hibernate.internal.FilterHelper;
import org.hibernate.internal.SessionFactoryImpl;
import org.hibernate.jdbc.ReturningWork;
import org.hibernate.jdbc.Work;
import org.hibernate.persister.entity.AbstractEntityPersister;
import org.hibernate.persister.entity.EntityPersister;
import org.hibernate.type.IntegerType;
import org.hibernate.type.Type;
import org.springframework.stereotype.Service;

import com.googlecode.cswish.annotation.Protected;
import com.googlecode.cswish.model.ModelInfo;
import com.googlecode.cswish.model.PageInfo;
import com.googlecode.cswish.model.PropertyInfo;
import com.googlecode.cswish.model.Reference;
import com.googlecode.cswish.oosearch.Condition;
import com.googlecode.cswish.oosearch.QLInfo;
import com.googlecode.cswish.oosearch.QLParse;
import com.googlecode.cswish.security.SafeManager;
import com.googlecode.cswish.struts.ContextHolder;
import com.googlecode.cswish.struts.hibernate.HibernateConfiguration;
import com.googlecode.cswish.struts.hibernate.IDynamicModel;
import com.googlecode.cswish.struts.hibernate.ehcache.AlwaysNotMatch;
import com.googlecode.cswish.struts.hibernate.ehcache.HibernateNameAttributeExtractor;
import com.googlecode.cswish.struts.hibernate.ehcache.IgnoreCaseLike;
import com.googlecode.cswish.struts.hibernate.ehcache.IgnoreTypeEqualTo;
import com.googlecode.cswish.struts.hibernate.ehcache.InSubCriteria;
import com.googlecode.cswish.struts.hibernate.ehcache.SiteCache;
import com.googlecode.cswish.util.ClassUtil;
import com.googlecode.cswish.util.Functions;
import com.googlecode.cswish.util.OgnlExUtil;
import com.googlecode.cswish.util.bean.ModelUtil;
import com.opensymphony.xwork2.ActionContext;

@Service
public class CacheSearchService {

	private static final Log logger = LogFactory.getLog(CacheSearchService.class);

	private final static int MIN_CHECK_PAGE = 5;
	private final static int MAX_PAGE_SIZE = 100;
	private final static int MAX_JOIN_SEARCH_ID_COUNT = 1000;
	static int CACHE_SEARCH_PERCENT = 20;			// (1 ~ 100)
	
	@Resource
	private GenericService genericService;

	@Resource
	private QLParse qlParse;
	
	@Resource
	private BeanIntrospector beanIntrospector;

	@PersistenceContext(unitName = "espirit")
	private EntityManager em;

	@Resource
	private HibernateConfiguration hibernateConfiguration;
	
	@Resource
	private SafeManager safeManager;
	
	@Resource
	private Functions functions;
	
	@Resource
	private OgnlExUtil ognlUtil;
	
	@Resource
	private ObjectFactory objectFactory;
	
	// store the Ehcache information
	private Map<String, CacheInfo> cacheInfos;
	
	@Resource
	private CacheService cacheService;
	
	@PostConstruct
	private void init() {
		cacheInfos = cacheService.newCache(CacheInfo.class);
	}
	
	private class CacheInfo {
		private int percentage;			// 0 ~ 100
		private int maxId;
		private int totalCount;
		private long lastModifiedTime;
	}
	
	private CacheManager cacheManager;
	
	private CacheManager getCacheManager() {
		if (cacheManager == null) {
			cacheManager = genericService.getCacheManager();
		}
		return cacheManager;
	}
	
	public void setCacheInfo(String modelName, Integer maxId, Integer percentage, Integer totalCount) {
		CacheInfo cacheInfo = cacheInfos.get(modelName);
		if (cacheInfo == null) {
			cacheInfo = new CacheInfo();
			cacheInfos.put(modelName, cacheInfo);
		}
		if (maxId != null) {
			cacheInfo.maxId = maxId;
		}
		if (percentage != null) {
			cacheInfo.percentage = percentage;
		}
		if (totalCount != null) {
			cacheInfo.totalCount = totalCount;
		}
		cacheInfo.lastModifiedTime = Timestamper.next();
	}
	public Integer checkAndGetPercentage(String modelName) {
		CacheInfo cacheInfo = cacheInfos.get(modelName);
		Integer percentage = cacheInfo == null ? null : cacheInfo.percentage;
		if (percentage != null && percentage == 100) {
			// check the entity cache
			SessionFactoryImplementor factory = (SessionFactoryImplementor)((Session)em.getDelegate()).getSessionFactory();
			Serializable[] spaces = factory.getEntityPersister(modelName).getPropertySpaces();
			Set<Serializable> spaceSet = new HashSet<Serializable>((int)(spaces.length * 4 / 3) + 1);
			for (Serializable space : spaces) {
				spaceSet.add(space);
			}
			
			//SessionImplementor session = (SessionImplementor)em.getDelegate();
			boolean isUpToDate = factory.getUpdateTimestampsCache().isUpToDate(spaceSet, 
					cacheInfo.lastModifiedTime);
			if (!isUpToDate) {
				if (logger.isDebugEnabled()) {
					logger.debug("Cache is expired, update cache:" + modelName);
				}
				
				// refresh cache
				ICacheSearchInitService initService = objectFactory.getBean(ICacheSearchInitService.class, false);
				initService.refreshCache(modelName);
				percentage = checkAndGetPercentage(modelName);
			}
		}
		return percentage;
	}
	public Integer getMaxCacheId(String modelName) {
		CacheInfo cacheInfo = cacheInfos.get(modelName);
		return cacheInfo == null ? null : cacheInfo.maxId;
	}
	public Integer getTotalCount(String modelName) {
		CacheInfo cacheInfo = cacheInfos.get(modelName);
		return cacheInfo == null ? null : cacheInfo.totalCount;
	}
	
	public boolean isDataCached(String modelName, boolean checkParent, int minPercentage) {
		Integer curPercentage = checkAndGetPercentage(modelName);
		if (curPercentage == null) {
			Class clazz = ClassUtil.getClass(modelName);
			Class checkClass = clazz.getSuperclass();
			while (checkClass != null && curPercentage == null) {
				curPercentage = checkAndGetPercentage(checkClass.getName());
				checkClass = checkClass.getSuperclass();
			}
		}
		
		if (curPercentage == null) {
			return false;
		} else {
			// TODO: maxRecordCount should be the db record count
			return curPercentage >= minPercentage ? true : false;
		}
	}
	
	private int getMaxCacheIdIfNotFullCache(String modelName, boolean exactResult) {
		if (!exactResult) {
			return 0;
		}
		
		Integer curPercentage = checkAndGetPercentage(modelName);
		if (curPercentage == null) {
			Class clazz = ClassUtil.getClass(modelName);
			Class checkClass = clazz.getSuperclass();
			while (checkClass != null && curPercentage == null) {
				modelName = checkClass.getName();
				curPercentage = checkAndGetPercentage(modelName);
				checkClass = checkClass.getSuperclass();
			}
		}
		
		if (curPercentage != 100) {
			return getMaxCacheId(modelName);
		} else {
			return 0;
		}
	}
	
	Ehcache getCache(String modelName) {
		Class clazz = ClassUtil.getClass(modelName);
		Ehcache cache = getCacheManager().getEhcache(clazz.getName());
		
		if (cache == null) {
			Class checkClazz = clazz.getSuperclass();
			
			while (cache == null && checkClazz != null) {
				cache = getCacheManager().getCache(checkClazz.getName());
				checkClazz = checkClazz.getSuperclass();
			}
		}
		
		if (cache instanceof SiteCache) {		// use original cache for better performance
			cache = ((SiteCache)cache).getSiteCache(ContextHolder.get().getSiteName());
		}
		return cache;
	}
	
	private List<Integer> simpleSearch(Object model, int start, int limit, String sort, String dir, boolean exactResult, boolean disablePermission) {
		StringBuilder sbModelName = new StringBuilder();
		Map<String, Object> searchModel = qlParse.toSearchModel(model, sbModelName);
		String modelName = sbModelName.toString();

		Class clazz = ClassUtil.getClass(modelName);
		Ehcache cache = getCache(modelName);
		if (cache == null) {
			logger.error(modelName + " is not cached class");
			return null;
		}
		
		boolean needFilterClass = !modelName.equals(cache.getName());
		Integer searchId = searchById(searchModel, cache);
		if (searchId != null) {
			List<Integer> ids = new ArrayList<Integer>(1);
			if (searchId >= 0) {
				ids.add(searchId);
			}
			return ids;
		}
		
		Query query = cache.createQuery();
		query.includeKeys();

		// filter class type: should the sub class
		List<Class> subClasses = beanIntrospector.getSubClass(clazz.getName());
		if (needFilterClass || subClasses != null && subClasses.size() > 0) {
			Attribute<String> attr = null;
			try {
				attr = cache.getSearchAttribute("class");
			} catch (CacheException ex) {
				attr = buidSearchAttribute(cache, null, "class");
			}
			// itself
			Criteria classCriteria = new IgnoreTypeEqualTo("class", clazz.getName());
			
			// sub class
			if (subClasses != null) {
				for (Class subClass : subClasses) {
					String className = subClass.getName();
					classCriteria = classCriteria.or(new IgnoreTypeEqualTo("class", className));
				}
			}
			query.addCriteria(classCriteria);
		}
		
		// exact result
		if (exactResult && !isDataCached(cache.getName(), false, 100)) {
			Attribute<Integer> attr = null;
			try {
				attr = cache.getSearchAttribute("id");
			} catch (CacheException ex) {
				attr = buidSearchAttribute(cache, clazz.getName(), "id");
			}
			Integer maxValidId = getMaxCacheId(cache.getName());
			attr.le(maxValidId);
		}
		
		// 1. build search condition
		if (searchModel.size() > 0) {
			Criteria criteria = null;
			criteria = buildSearchCriteria(searchModel, modelName, "", cache, criteria);
			if (criteria != null) {
				query.addCriteria(criteria);
			}
		}
		
		// check permission
		if (!disablePermission) {
			RecordPermissionCriteria permissionCriteria = new RecordPermissionCriteria(safeManager, functions, modelName, "search");
			if (!permissionCriteria.isAlwaysTrue()) {
				query.addCriteria(permissionCriteria);
			}
		}
		
		// 2. search & pagination
		PageInfo pageInfo = new PageInfo();
		pageInfo.setPageSize(limit);
		pageInfo.setPageNoByRowNo(start);
		if (start >= 0 && limit > 0) {
			int pageSize = pageInfo.getPageSize();
			if (pageSize > MAX_PAGE_SIZE) {
				pageSize = MAX_PAGE_SIZE;
			}
			int pageNo = pageInfo.getPageNo();
			int maxResults = (pageNo < MIN_CHECK_PAGE ? MIN_CHECK_PAGE : pageNo + 1) * pageSize;
			if (limit < maxResults) {
				query.maxResults(maxResults);
			} else {
				query.maxResults(limit);
			}
		}
		if (StringUtils.isNotEmpty(sort)) {
			if (sort.startsWith("a.")) {
				sort = sort.substring(2);
			}
			Attribute searchAttribute;
			try {
				searchAttribute = cache.getSearchAttribute(sort);
			} catch (CacheException ex) {
				searchAttribute = buidSearchAttribute(cache, modelName, sort);
			}
			Direction direction = "ASC".equals(dir) || "asc".equals(dir) ? Direction.ASCENDING : Direction.DESCENDING;
			query.addOrderBy(searchAttribute, direction);
		}
		Results results = query.execute();
		List<Result> pageResults = results.range(start, pageInfo.getPageSize());

		// 3. convert result to hibernate model
		List<Integer> ids = new ArrayList<Integer>();
		for (Result result : pageResults) {
			CacheKey cacheKey = (CacheKey) result.getKey();
			ids.add((Integer)cacheKey.getKey());
		}
		return ids;
	}

	Integer searchById(Map<String, Object> searchModel, Ehcache cache) {
		Object idSearchValue = searchModel.get("id");
		boolean searchById = false;
		if (notEmpty(idSearchValue)) {
			char firstChar = idSearchValue.toString().charAt(0);
			if (firstChar >= '0' && firstChar <= '9') {
				searchById = true;
			}
		}
		
		Integer foundId = null;
		if (searchById) {
			String modelName = cache.getName();
			Integer id = ognlUtil.convertValue(searchModel.get("id"), Integer.class);
			SessionFactoryImpl factory = (SessionFactoryImpl)hibernateConfiguration.getSessionFactory();
			CacheKey cacheKey = new CacheKey(id, IntegerType.INSTANCE, modelName, null, factory);
			if (cache.get(cacheKey) != null) {
				foundId = id;
			} else {
				foundId = -1;		// it means false because the id can't be negative
			}
		} else {
			foundId = null;
		}
		return foundId;
	}

	private Criteria buildSearchCriteria(Map<String, Object> searchModel, String modelName, String namePrefix, Ehcache cache, Criteria criteria) {
		for (Entry<String, Object> entry : searchModel.entrySet()) {
			String property = entry.getKey();
			Object searchValue = entry.getValue();
			if (notEmpty(searchValue) && !property.equals(IDynamicModel.ENTITY_NAME)) {
				Attribute<String> attr = null;
				try {
					attr = cache.getSearchAttribute(property);
				} catch (CacheException ex) {
					attr = buidSearchAttribute(cache, modelName, property);
				}
				
				if (attr != null) {
					if (searchValue instanceof Map) {
						Map<String, Object> subSearchModel = (Map<String, Object>)searchValue;
						String subModelName = (String)subSearchModel.get(IDynamicModel.ENTITY_NAME);
						boolean isEmbeddable = ClassUtil.getClass(subModelName).getAnnotation(Embeddable.class) != null;
						if (isEmbeddable) {
							buildSearchCriteria(subSearchModel, modelName, namePrefix + property + '.', cache, criteria);
						} else {
							Ehcache subCache = getCache(subModelName);
							// quick check for id search
							Integer searchId = searchById(subSearchModel, subCache);
							
							Criteria inSub = null;
							if (searchId != null) {
								if (searchId >= 0) {
									inSub = new IgnoreTypeEqualTo(property, searchId.toString());
								} else {
									inSub = new AlwaysNotMatch();
								}
							} else {
								Criteria subCriteria = buildSearchCriteria(subSearchModel, subModelName, namePrefix, subCache, null);
								if (subCriteria != null) {
									inSub = new InSubCriteria(property, subCache, subCriteria);
								}
							}
							if (inSub != null) {
								if (criteria == null) {
									criteria = inSub;
								} else {
									criteria.and(inSub);
								}
							}
						}
					} else if (searchValue instanceof Collection) {
						// TODO:
					} else {
						PropertyInfo propInfo = beanIntrospector.getPropertyInfo(modelName, property);
						criteria = appendCriteria(criteria, searchValue.toString(), attr, propInfo);
					}
				}
			}
		}
		return criteria;
	}

	private <T> Attribute<T> buidSearchAttribute(Ehcache cache, String modelName, String nestedProperty) {
		Attribute<T> attr = null;
		try {
			// build the AttributeExtractor by lazy model
			boolean passCheck = "class".equals(nestedProperty) || 
				beanIntrospector.getPropertyInfo(modelName, nestedProperty) != null;
			if (passCheck) {
				Field field = net.sf.ehcache.Cache.class.getDeclaredField("compoundStore");
				field.setAccessible(true);
				Store store = (Store)field.get(cache);
				Map<String, AttributeExtractor> extractors = new HashMap<String, AttributeExtractor>(2);
				extractors.put(nestedProperty, new HibernateNameAttributeExtractor());
				store.setAttributeExtractors(extractors);
				attr = cache.getSearchAttribute(nestedProperty);
			} else {
				if (logger.isWarnEnabled()) {
					logger.warn(nestedProperty + " is not searchable property, skip it");
				}
			}
		} catch (Exception ex) {								
			if (logger.isWarnEnabled()) {
				logger.warn(nestedProperty + " is not searchable property, skip it", ex);
			}
		}
		return attr;
	}

	void createTmpTable(Connection connection, String tableName) throws SQLException {
		Statement stmtCreate = connection.createStatement();
		try {
			stmtCreate.execute("CREATE TEMPORARY TABLE IF NOT EXISTS " + tableName + "(id INTEGER) TYPE = HEAP");
		} finally {
			if (stmtCreate != null) {
				stmtCreate.close();
			}
		}
	}
	void dropTmpTable(final List<String> tableNames) throws SQLException {
		Session session = (Session) em.getDelegate();
		session.doWork(new Work() {
			@Override
			public void execute(Connection connection) throws SQLException {
				Statement stmtDelete = connection.createStatement();
				try {
					for (String tableName : tableNames) {
						stmtDelete.execute("DROP TEMPORARY TABLE IF EXISTS " + tableName);
					}
				} finally {
					if (stmtDelete != null) {
						stmtDelete.close();
					}
				}
			}
		});
	}
	void insertTmpData(final String tableName, final List<Integer> ids) {
		Session session = (Session) em.getDelegate();
		session.doWork(new Work() {
			@Override
			public void execute(Connection connection) throws SQLException {
				createTmpTable(connection, tableName);
				
				String sql = "insert into " + tableName + " (id) values(?)";
				connection.setAutoCommit(false);
				PreparedStatement pstmtInsert = connection.prepareStatement(sql);
				try {
					// Insert 10 rows of data
					for (Integer id : ids) {
						pstmtInsert.setInt(1, id);
						pstmtInsert.addBatch();
					}
					// Execute the batch
					pstmtInsert.executeBatch();
					connection.commit();
				} finally {
					if (pstmtInsert != null) {
						pstmtInsert.close();
					}
				}
			}
		});
	}

	int checkTmpDataCount(final String tableName) {
		Session session = (Session) em.getDelegate();
		return session.doReturningWork(new ReturningWork<Integer>() {
		
			@Override
			public Integer execute(Connection connection) throws SQLException {
				
				String sql = "select count(*) from " + tableName;
				PreparedStatement stmt = connection.prepareStatement(sql);
				try {
					stmt.execute(sql);
					ResultSet rs = stmt.getResultSet();
					rs.next();
					int count = rs.getInt(1);
					return count;
				} finally {
					if (stmt != null) {
						stmt.close();
					}
				}
			}
		});
	}
	
	private Criteria appendCriteria(Criteria criteria, String strSearchValue, Attribute<String> attr, PropertyInfo propInfo) {
		// first char is join flag
		char joinFlag = strSearchValue.charAt(0);
		if (joinFlag == '|') { // TODO: QLParse.getCmpFlag
			strSearchValue = strSearchValue.substring(1);
		}
		char cmpFlag = strSearchValue.charAt(0);
		if (cmpFlag == '>' || cmpFlag == '<' || cmpFlag == '!' || cmpFlag == '=' || cmpFlag == '^') {
			strSearchValue = strSearchValue.substring(1);
		} else {
			cmpFlag = '\0';
			if (propInfo != null) {
				Class propType = propInfo.getTypeClass();
				if (Number.class.isAssignableFrom(propType) || Date.class.isAssignableFrom(propType)) {
					cmpFlag = '=';
				}
			}
		}

		Criteria attCriteria = null;
		if (cmpFlag == '\0') {
			IgnoreCaseLike like = new IgnoreCaseLike(attr.getAttributeName(), "*" + strSearchValue + "*");
			attCriteria = like;
		} else if (cmpFlag == '>') {
			attCriteria = attr.gt(strSearchValue);
		} else if (cmpFlag == '<') {
			attCriteria = attr.le(strSearchValue);
		} else if (cmpFlag == '!') {
			attCriteria = attr.ne(strSearchValue);
		} else if (cmpFlag == '=') {
			IgnoreTypeEqualTo eq = new IgnoreTypeEqualTo(attr.getAttributeName(), strSearchValue);
			attCriteria = eq;
		} else if (cmpFlag == '^') {
			IgnoreCaseLike like = new IgnoreCaseLike(attr.getAttributeName(), strSearchValue + "*");
			attCriteria = like;
		}

		if (attCriteria != null) {
			if (criteria == null) {
				criteria = attCriteria;
			} else {
				if (joinFlag == '|') {
					criteria = criteria.or(attCriteria);
				} else {
					criteria = criteria.and(attCriteria);
				}
			}
		}
		return criteria;
	}
	
	class CacheModelInfo {
		
		private String nestedProperty;
		
		private Map<String, Object> searchModel;
		
		private List<Integer> ids;
		
		private int maxValidId;
		
		public CacheModelInfo(Map<String, Object> searchModel, String nestedProperty, List<Integer> ids, int maxValidId) {
			this.searchModel = searchModel;
			this.nestedProperty = nestedProperty;
			this.ids = ids;
			this.maxValidId = maxValidId;
		}
		
		public String toString() {
			String modelName = (String)searchModel.get(IDynamicModel.ENTITY_NAME);
			return modelName + '(' + nestedProperty + ")," + maxValidId;
		}
	}
	
	public <T> PageInfo<T> search(T model, int start, int limit,
    		String sort, String dir, boolean retExactResult, boolean newSearch, Reference<QLInfo> qlInfoRef) {
    	return search(model, start, limit, sort, dir, retExactResult, newSearch, qlInfoRef, null, true, false);
    }
	
	@Protected
	public <T> PageInfo<T> search(T model, int start, int limit, String sort, String dir, int defaultCount, boolean retExactResult) {
		boolean calculateCount = defaultCount == -1;
    	PageInfo<T> pageInfo = search(model, start, limit, sort, dir, retExactResult,
    			true, null, null, calculateCount, false);
    	if (!calculateCount) {
    		pageInfo.setTotalCount(defaultCount);
    	}
    	return pageInfo;
	}
	
	
	public <T> PageInfo<T> search(T model, int start, int limit, String sort, String dir, boolean exactResult,
	    		boolean newSearch, Reference<QLInfo> qlInfoRef, SearchCallback callback, boolean calculateCount, boolean disablePermission) {
		String modelSessionKey = null;
		boolean foundCacheModel = false;
		if (!newSearch) {
			// try to load the search condition from session
			ActionContext actionContext = ActionContext.getContext();
			String actionName = (String) actionContext.getActionInvocation().getStack().getContext().get("actionName");
			modelSessionKey = actionName + "@model";
			Object oldModel = actionContext.getSession().get(modelSessionKey);
			if (oldModel != null) {
				model = (T) oldModel;
				foundCacheModel = true;
			}
		}
		
		PageInfo<T> pageInfo = null;
//		if (newSearch || foundCacheModel) {
//			// try to load from cache
//			pageInfo = searchByCache(model, start, limit, sort, dir, exactResult, newSearch, qlInfoRef, callback, calculateCount, disablePermission);
//		}
		
		if (pageInfo == null) {					// no cache model, call search method directly
			if (logger.isDebugEnabled()) {
				logger.debug("Cache doesn't hit, load data from db: " + ClassUtil.getClassName(model));
			}
			pageInfo = (PageInfo<T>)genericService.search(model, start, limit, sort, dir, newSearch, qlInfoRef, callback, calculateCount, disablePermission);
		}
		return pageInfo;
	}
	
	private void addFilterDefinition(SessionFactory factory, String filterName,
				FilterDefinition filterDefinition) {
		try {
			Field field = factory.getClass().getDeclaredField("filters");
			field.setAccessible(true);
			Map<String, FilterDefinition> filters = (Map<String, FilterDefinition>)field.get(factory);
			filters.put(filterName, filterDefinition);
		} catch (Exception ex) {
			logger.error("add definition", ex);
		}
	}
	
	private void rebuildFilterHelper(String modelName, List<FilterDefinition> filters) {
		SessionFactoryImpl factory = (SessionFactoryImpl)hibernateConfiguration.getSessionFactory();
		Map<String, EntityPersister> entityPersisters = factory.getEntityPersisters();
		EntityPersister persister = entityPersisters.get(modelName);
		List<FilterConfiguration> filterCfgs = new ArrayList<FilterConfiguration>(filters.size());
		for (FilterDefinition def : filters) {
			FilterConfiguration cfg = new FilterConfiguration(def.getFilterName(), def.getDefaultFilterCondition(), true, 
					new HashMap<String, String>(), new HashMap<String, String>(), hibernateConfiguration.getClassMapping(modelName));
			filterCfgs.add(cfg);
		}
		FilterHelper filterHelper = new FilterHelper(filterCfgs, factory);
		try {
			Field field = AbstractEntityPersister.class.getDeclaredField("filterHelper");
			field.setAccessible(true);
			field.set(persister, filterHelper);
		} catch (Exception ex) {
			logger.error("add filter helper", ex);
		}
	}
	
	private boolean enableSimpleSearch(Map<String, Object> searchModel, boolean exactResult) {
		boolean enableSimpleSearch = true;
		for (Object searchValue : searchModel.values()) {
			if (searchValue instanceof Map) {
				Map<String, Object> subModel = (Map<String, Object>)searchValue;
				String subType = (String)subModel.get(IDynamicModel.ENTITY_NAME);
				// TODO: supports partition query for the cache property of cache object? 100 --> CACHE_SEARCH_PERCENT
				enableSimpleSearch = isDataCached(subType, true, 100)
						&& enableSimpleSearch(subModel, exactResult);
			} else if (searchValue instanceof Collection) {
				enableSimpleSearch = false;
			}
		}
		return enableSimpleSearch;
	}
	private boolean enableSimpleSearch(String modelName, String sortProperty, boolean exactResult) {
		if (sortProperty == null || sortProperty.length() == 0) {
			return true;
		}
		
		if (sortProperty.startsWith("a.")) {
			sortProperty = sortProperty.substring(2);
		}
		ModelInfo modelInfo = beanIntrospector.getModelInfo(modelName);
		boolean enableSimpleSearch = true;
		String[] properties = StringUtils.splitByWholeSeparator(sortProperty, ".");
		for (int i = 0, len = properties.length - 1; i < len; i++) {
			String prop = properties[i];
			PropertyInfo propertyInfo = modelInfo.getProperty(prop);
			String subType = propertyInfo.getType();
			
			boolean isEmbed = false;
			Method method = propertyInfo.getReadMethod();
			if (method != null) {
				Embedded embed = method.getAnnotation(Embedded.class);
				isEmbed = embed != null;
			}
			if (!isEmbed) {
				if (!isDataCached(subType, true, CACHE_SEARCH_PERCENT)) {
					enableSimpleSearch = false;
					break;
				}
			}
			modelInfo = beanIntrospector.getModelInfo(subType);
		}
		return enableSimpleSearch;
	}

	private void populateCacheSearchModel(String nestedProperty, Map<String, Object> searchModel, String sort, String dir,
			boolean exactResult, Map<Object, CacheModelInfo> cacheModelIdList, boolean disablePermission) {
		String type = (String)searchModel.get(IDynamicModel.ENTITY_NAME);
		boolean isEmbeddable = ClassUtil.getClass(type).getAnnotation(Embeddable.class) != null;
		if (isEmbeddable) {		// don't check the embeddable model, it belongs to the parent class
			return;
		}
		
		int size = size(cacheModelIdList);
		if (isDataCached(type, true, exactResult ? CACHE_SEARCH_PERCENT : 0)) {
			// TODO: support the nested property at future?
			if (nestedProperty == null || nestedProperty.indexOf('.') == -1) {
				boolean enableSimpleSearch = enableSimpleSearch(searchModel, exactResult) 
						&& enableSimpleSearch(type, sort, exactResult);
				if (enableSimpleSearch) {
					
					// search & populate the id list
					int limit = MAX_JOIN_SEARCH_ID_COUNT - size;
					
					int maxCacheId = getMaxCacheIdIfNotFullCache(type, exactResult);
					List<Integer> matchedIds = simpleSearch(searchModel, 0, limit, sort, dir, exactResult, disablePermission);
					cacheModelIdList.put(searchModel, new CacheModelInfo(searchModel, nestedProperty, matchedIds, maxCacheId));
					return;
				}
			}
		}
		
		if (size < MAX_JOIN_SEARCH_ID_COUNT) {
			for (Entry<String, Object> entry : searchModel.entrySet()) {
				Object value = entry.getValue();
				if (value instanceof Collection) {
					// FIXME: correct?
					Collection list = (Collection)value;
					String newNestedProperty = getNestedProperty(nestedProperty, entry.getKey());
					for (Object val : list) {
						if (val instanceof Map) {
							populateCacheSearchModel(newNestedProperty,
									(Map<String, Object>)val, sort, dir, exactResult, cacheModelIdList, disablePermission);
						}
					}
				} else if (value instanceof Map) {
					String newNestedProperty = getNestedProperty(nestedProperty, entry.getKey());
					populateCacheSearchModel(newNestedProperty,
							(Map<String, Object>)value, sort, dir, exactResult, cacheModelIdList, disablePermission);
				} else {
					// do nothing
				}
			}
		}
	}
	
	@SuppressWarnings("unchecked")
	private <T> PageInfo<T> searchByCache(T model, int start, int limit, String sort, String dir, boolean exactResult, 
			boolean newSearch, Reference<QLInfo> qlInfoRef, SearchCallback callback, boolean calculateCount, boolean disablePermission) {
		if (callback != null) {
			boolean enableCacheSearch = true;
			try {
				QLInfo qlInfo = new QLInfo();
				qlInfo.setQl("");
				qlInfo.setDir(dir);
				String modelName = ClassUtil.getClassName(model);
				Map<String, Object> parameterModel = null;
				qlParse.setSort(qlInfo, modelName, parameterModel, sort);
				qlInfo = callback.afterBuildQL(newSearch, qlInfo);
				
				// cache search doesn't support the ql at present
				enableCacheSearch = qlInfo.getQl().isEmpty();
				
				if (enableCacheSearch) {
					String sortAndDir = qlInfo.getSortAndDir();
					if (sortAndDir.length() > 0) {
						String[] sortInfos = StringUtils.splitByWholeSeparator(sortAndDir, ",");
						String[] firstSortAndDir = StringUtils.splitByWholeSeparator(sortInfos[0], " ");
						sort = firstSortAndDir[0];
						dir = firstSortAndDir[1];
					}
				}
			} catch (Exception ex) {
				logger.debug("Failed to execute the callback, stop the cache search", ex);
				enableCacheSearch = false;
			}
			
			if (!enableCacheSearch) {
				return null;
			}
		}
		
		
		StringBuilder sbModelName = new StringBuilder();
		Map<String, Object> searchModel = qlParse.toSearchModel(model, sbModelName);
		String modelName = sbModelName.toString();
		Map<Object, CacheModelInfo> cacheModelInfos = new IdentityHashMap<Object, CacheModelInfo>();
		
		// get the matched list
		populateCacheSearchModel(null, searchModel, sort, dir, exactResult, cacheModelInfos, disablePermission);
		
		PageInfo<T> pageInfo = null;
		int size = size(cacheModelInfos);
		if (size > 0 && size < MAX_JOIN_SEARCH_ID_COUNT) {
			if (logger.isDebugEnabled()) {
				logger.debug("search " + modelName + ", Load data from cache, cache hits " + size);
			}
			
			// adjust the search model & remove the found search condition
			Map<String, Object> newSearchModel = rebuildSearchModel(searchModel, cacheModelInfos);
			if (newSearchModel == null) {	// all data exists in the cache model
				Class clazz = ClassUtil.getClass(modelName);
				List ids = null;
				for (CacheModelInfo cacheModelInfo : cacheModelInfos.values()) {
					if (cacheModelInfo.searchModel.get(IDynamicModel.ENTITY_NAME).equals(modelName)) {
						ids = cacheModelInfo.ids;
					}
				}
				
				// pagination
				int minTotalCount = ids.size();
				List paginationIds;
				if (start >= 0 && limit > 0) {
					int toIndex = start + limit;
					if (toIndex > ids.size()) {
						toIndex = ids.size();
					}
					paginationIds = ids.subList(start, toIndex);
				} else {
					paginationIds = ids;
				}
				
				// load data
				List<T> data = genericService.load(clazz, paginationIds, true);
				boolean isExactResult = tryMakupMissData(data, searchModel, start, limit, minTotalCount, sort, dir);
				// increase the total count
				minTotalCount += data.size() - paginationIds.size();
				
				// build PageInfo
				pageInfo = new PageInfo();
				pageInfo.setPageSize(limit);
				pageInfo.setPageNoByRowNo(start);
				pageInfo.setTotalCount(minTotalCount);
				pageInfo.setData(data);
				if (!isExactResult) {
					pageInfo.setDesc(">");
				}
			} else {
				// backup the old search model
				if (newSearch) {
					ActionContext actionContext = ActionContext.getContext();
					if (actionContext != null) {
						String actionName = (String) actionContext.getActionInvocation().getStack().getContext().get("actionName");
						String modelSessionKey = actionName + "@model";
						actionContext.getSession().put(modelSessionKey, searchModel);
					}
				}
				searchModel = newSearchModel;
				
				// searchModel restores at the begin, so treat every search operation as new operation
				if (size <= 10) {
					pageInfo = searchByJoinSql(searchModel, cacheModelInfos.values(), start, limit, sort, dir, true,
							qlInfoRef, calculateCount, disablePermission);
				} else {
					pageInfo = searchByJoinTable(searchModel, cacheModelInfos.values(), start, limit, sort, dir, true,
						qlInfoRef, calculateCount, disablePermission);
				}
				// TODO: load more data if the cache doesn't contains full data? 
			}
		} else if (cacheModelInfos.size() > 0) {		// found the cache model but the result is zero
			pageInfo = new PageInfo<T>(limit);
    		pageInfo.setData(new ArrayList<T>(0));
		}
		return pageInfo;
	}

	private PageInfo searchByJoinSql(Map<String, Object> searchModel, final Collection<CacheModelInfo> cacheModelIdInfo, int start, int limit, String sort,
			String dir, boolean newSearch, Reference<QLInfo> qlInfoRef, boolean calculateCount, boolean disablePermission) {
		SearchCallback callback = null;
		if (newSearch || qlInfoRef == null) {
			callback = new SearchCallback() {
				@Override
				public QLInfo afterBuildQL(boolean newSearch, QLInfo qlInfo) {
					if (!newSearch) {
						return qlInfo;
					}
					// convert idList as additional sql
					StringBuilder additionalQL = new StringBuilder();
					List<Object> params = new ArrayList<Object>();
					additionalQL.append("(");
					for (CacheModelInfo modelInfo : cacheModelIdInfo) {
						if (additionalQL.length() > 1) {
							additionalQL.append(" and ");
						}
						additionalQL.append("(");
						boolean firstId = true;
						for (Integer id : modelInfo.ids) {
							if (firstId) {
								firstId = false;
							} else {
								additionalQL.append(" or ");
							}
							additionalQL.append(QLParse.DEFAULT_ALIAS);
							if (modelInfo.nestedProperty != null && modelInfo.nestedProperty.length() > 0) {
								additionalQL.append('.').append(modelInfo.nestedProperty);
							}
							additionalQL.append('.').append("id=?");
							params.add(id);
						}
						// build the db condition
						// id >= ? and original where condition
						if (modelInfo.maxValidId != 0) {
							additionalQL.append(" or ");
							String tableAlias;  
							if (modelInfo.nestedProperty != null && modelInfo.nestedProperty.length() > 0) {
								tableAlias = QLParse.DEFAULT_ALIAS + '.' + modelInfo.nestedProperty;
							} else {
								tableAlias = QLParse.DEFAULT_ALIAS;
							}
							additionalQL.append(tableAlias);
							additionalQL.append('.').append("id>?");
							params.add(modelInfo.maxValidId);
							
							QLInfo where = qlParse.getWhereClause(tableAlias, modelInfo.searchModel, Condition.SEARCH_LIKEFIRST);
							additionalQL.append(" and (").append(where.getQl()).append(")");
							params.addAll(where.getParams());
						}
						additionalQL.append(")");
					}
					additionalQL.append(")");
					if (additionalQL.length() <= 2) { 		// equals "()"
						additionalQL = new StringBuilder("1<>1");
					}
					
					int index = qlInfo.getQl().indexOf(" where ");
					String ql;
					if (index > 0) {
						ql = qlInfo.getQl() + " and " + additionalQL;
					} else {
						ql = qlInfo.getQl() + " where " + additionalQL;
					}
					qlInfo.setQl(ql);
					if (qlInfo.getParams() != null) {
						qlInfo.getParams().addAll(params);
					} else {
						qlInfo.setParams(params);
					}
					return qlInfo;
				}
			};
		}
		PageInfo pageInfo = genericService.search(searchModel, start, limit, sort, dir,
				newSearch, qlInfoRef, callback, calculateCount, disablePermission);
		return pageInfo;
	}

	private PageInfo searchByJoinTable(Map<String, Object> searchModel, Collection<CacheModelInfo> cacheModelInfos, int start, int limit,
			String sort, String dir, boolean newSearch,
			Reference<QLInfo> qlInfoRef, boolean calculateCount, boolean disablePermission) {
		// prepare the temporary data
		List<String> tmpTableNames = new ArrayList<String>();
		for (CacheModelInfo cacheModelInfo : cacheModelInfos) {
			String theModelName = (String)cacheModelInfo.searchModel.get(IDynamicModel.ENTITY_NAME);
			String tmpTableName = theModelName.substring(theModelName.lastIndexOf('.') + 1) + "Tmp";
			tmpTableNames.add(tmpTableName);
			insertTmpData(tmpTableName, cacheModelInfo.ids);
		}
		
		// prepare the filter condition
		EventSource session = (EventSource) em.getDelegate();
		List<FilterDefinition> filters = new ArrayList<FilterDefinition>();
		for (CacheModelInfo cacheModelInfo : cacheModelInfos) {
			String theModelName = (String)cacheModelInfo.searchModel.get(IDynamicModel.ENTITY_NAME);
			boolean withMaxId = cacheModelInfo.maxValidId != 0;
			QLInfo sqlInfo = null;
			if (withMaxId) {
				ModelInfo modelInfo = beanIntrospector.getModelInfo(theModelName);
				String tableName = modelInfo.getTable();
				sqlInfo = qlParse.getSQLWhereClause(tableName, cacheModelInfo.searchModel, Condition.SEARCH_LIKEFIRST);
			}
			String filterName = getAndCreateFilter(cacheModelInfo, sqlInfo, filters);
			
			Filter filter = session.enableFilter(filterName);
			if (withMaxId) {
				if (logger.isDebugEnabled()) {
					logger.debug("search temp table + with max valid id:" + cacheModelInfo.maxValidId);
				}
				filter.setParameter("maxValidId", cacheModelInfo.maxValidId);
				List<String> paramNames = sqlInfo.getParamNames();
				List<Object> paramValues = sqlInfo.getParams();
				for (int i = 0, len = sqlInfo.getParamNames().size(); i < len; i++) {
					filter.setParameter(paramNames.get(i), paramValues.get(i));
				}
			}
		}
		if (filters.size() > 0) {
			String modelName = (String)searchModel.get(IDynamicModel.ENTITY_NAME);
			rebuildFilterHelper(modelName, filters);
		}
		
		// search data
		PageInfo pageInfo = genericService.search(searchModel, start, limit, sort, dir,
				newSearch, qlInfoRef, null, calculateCount, disablePermission);
		
		// remove the temporary data
		try {
			dropTmpTable(tmpTableNames);
		} catch (Exception ex) {
			logger.warn("Failed to clean the temporary data", ex);
		}
		return pageInfo;
	}
	
	private String getAndCreateFilter(CacheModelInfo cacheModelInfo, QLInfo sqlInfo, List<FilterDefinition> filters) {
		String modelName = (String)cacheModelInfo.searchModel.get(IDynamicModel.ENTITY_NAME);
		SessionFactory factory = hibernateConfiguration.getSessionFactory();
		String tmpTableName = modelName.substring(modelName.lastIndexOf('.') + 1) + "Tmp";
		String filterCondition = cacheModelInfo.nestedProperty + "_id in (select " + tmpTableName 
				+ ".id from " + tmpTableName +")";
		Map<String, Type> parameterTypes = new HashMap<String, Type>();
		boolean withMaxId = cacheModelInfo.maxValidId != 0;
		String filterName = null;
		if (withMaxId) {
			ModelInfo modelInfo = beanIntrospector.getModelInfo(modelName);
			String tableName = modelInfo.getTable();
			String columnName = cacheModelInfo.nestedProperty + "_id";
			StringBuilder additionParam = new StringBuilder();
			additionParam.append(" or ").append(columnName).append(">=:maxValidId");
			additionParam.append(" and ").append(columnName).append(" in ");
			additionParam.append("(select id from ").append(tableName).append(" where ").append(sqlInfo.getQl()).append(")");
			filterCondition = filterCondition + additionParam;
			
			Set filterNames = factory.getDefinedFilterNames();
			for (Object nameObj : filterNames) {
				String name = (String)nameObj;
				FilterDefinition filterDefinition = factory.getFilterDefinition(name);
				String condition = filterDefinition.getDefaultFilterCondition();
				if (condition.equals(filterCondition)) {
					filterName = name;
					break;
				}
			}
			
			if (filterName == null) {
				filterName = "filter" + tmpTableName + filterNames.size();
				for (String name : sqlInfo.getParamNames()) {
					PropertyInfo propInfo = modelInfo.getProperty(name);
					String shortName = propInfo.getType().substring(propInfo.getType().lastIndexOf('.') + 1);
					String typeName = "org.hibernate.type." + shortName + "Type";
					Type type = (Type)ModelUtil.newInstance(ClassUtil.getClass(typeName));
					parameterTypes.put(name, type);
				}
			}
		} else {
			filterName = "filter" + tmpTableName;
		}
		
		if (!factory.getDefinedFilterNames().contains(filterName)) {
			FilterDefinition filterDefinition = new FilterDefinition(filterName, 
					filterCondition, parameterTypes);
			addFilterDefinition(factory, filterName, filterDefinition);
			filters.add(filterDefinition);
		}
		return filterName;
	}
	
	private boolean tryMakupMissData(List data, Object model, int start, int limit, int totalSize, String sort, String dir) {
		String modelName = ClassUtil.getClassName(model);
		boolean allDataCached = isDataCached(modelName, true, 100);
		boolean isExactResult = allDataCached;
		if (!allDataCached) {
			int dbStart = start + data.size();
			int missPageDataCount = limit - data.size();
			boolean isLastPage = totalSize - limit <= start;
			if (missPageDataCount > 0 || isLastPage) {
				// NOTICE: the pre-condition is: the order of cache element is same with db element
				List dbData;
				// FIXME: reload all the missed data to ensure the correct data?
				boolean reloadAllData = limit < 100;
				if (reloadAllData) {
					if (logger.isDebugEnabled()) {
						logger.debug("Ignore the cache result, reload from db:" + modelName);
					}
					dbData = genericService.search(model, start, limit + 1, sort, dir, false).getData();
					data.clear();
					if (dbData.size() > limit) {
						data.addAll(dbData.subList(0, limit));
						isExactResult = false;
					} else {
						data.addAll(dbData);
						isExactResult = true;
					}
				} else {
					if (logger.isDebugEnabled()) {
						logger.debug("Ignore the cache result, reload parts of data from db:" + modelName);
					}
					dbData = genericService.search(model, dbStart, missPageDataCount + 1, sort, dir, false).getData();
					if (dbData.size() > missPageDataCount) {
						data.addAll(dbData.subList(0, missPageDataCount));
						isExactResult = false;
					} else {
						data.addAll(dbData);
						isExactResult = true;
					}
				}
			}
		}
		return isExactResult;
	}

	private String getNestedProperty(String parent, String child) {
		if (parent == null) {
			return child;
		} else {
			return parent + '.' + child;
		}
	}
	
	private Map<String, Object> rebuildSearchModel(Map<String, Object> searchModel, Map<Object, CacheModelInfo>  cacheModelInfo) {
		// TODO: any problem if the cache model references the normal model and the normal exists filter condition?
		CacheModelInfo searchModelInfo = cacheModelInfo.get(searchModel);
		if (searchModelInfo != null) {
				return null;
		}
		
		Map<String, Object> newModel = new HashMap<String, Object>(searchModel.size() * 4 / 3);
		for (Entry<String, Object> entry : searchModel.entrySet()) {
			Object value = entry.getValue();
			Object newValue = null;
			if (value instanceof Collection) {
				Collection list = (Collection)value;
				List newList = new ArrayList(list.size());
				for (Object val : list) {
					if (val instanceof Map) {
						Map<String, Object> newListVal = rebuildSearchModel((Map<String, Object>)val, cacheModelInfo);
						
						if (newListVal != null) {
							newList.add(newListVal);
						}
					}
				}
				if (newList.size() > 0) {
					newValue = newList;
				}
			} else if (value instanceof Map) {
				newValue = rebuildSearchModel((Map<String, Object>)value, cacheModelInfo);
			} else {
				newValue = value;
			}
			
			if (newValue != null) {
				newModel.put(entry.getKey(), newValue);
			}
		}
		return newModel;
	}
	
	private int size(Map<Object, CacheModelInfo> cacheModelIdList) {
		int size = 0;
		for (CacheModelInfo idInfo : cacheModelIdList.values()) {
			size += idInfo.ids.size();
		}
		if (size == 0 && cacheModelIdList.size() > 0) {
			size = 1;		// use '1' to identify the cache query executed
		}
		return size;
	}
	
	private boolean notEmpty(Object value) {
    	boolean hasValue = false;
    	
    	if (value != null) {
    		Class type = value.getClass();
    		if (type != null && type.isPrimitive()) {
        		if (boolean.class == type) {
        			hasValue = ((Boolean)value).booleanValue();
        		} else {
        			int i = Integer.parseInt(value.toString());
        			hasValue = (i != 0);
        		}
        	} else {
	    		if (value instanceof Collection) {
	    			try {
		    			if (((Collection)value).size() > 0) {//avoid the hibernate lazy load exception
		    				hasValue = true;
		    			}
	    			} catch (Throwable t) {
	    			}
	    		} else if (value instanceof String) {
	    			hasValue = ((String) value).length() > 0;
	    		} else if (value instanceof Map) {
	    			hasValue = ((Map)value).size() > 1;		// skip IDynamicModel.ENTITY_NAME
	    		} else {
	    			hasValue = true;
	    		}
        	}
    	}
		return hasValue;
	}
}
